package com.woldier.filesystem.service.impl;


import com.woldier.base.exception.CommonError;
import com.woldier.base.exception.DoopfsException;
import com.woldier.filesystem.service.HdfsService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.stereotype.Service;
import java.io.File;
import java.io.IOException;

/**
 * @author woldier
 * @version 1.0
 * @description Hdfs api 实现
 * @date 2023/3/23 15:10
 **/
@Service
@RequiredArgsConstructor
@Slf4j
public class HdfsServiceImpl implements HdfsService {
    private final FileSystem fileSystem;
    @Override
    public void mkDir(String path) {
        mkDir(new Path(path));
    }

    @Override
    public void mkDir(Path path) {

    }

    @Override
    public boolean delete(String path) throws DoopfsException {
        return delete(new Path(path));
    }

    @Override
    public boolean delete(Path path) throws DoopfsException {
        boolean tag = false ;
        try {
            tag = fileSystem.delete(path,true);
        } catch (IOException e) {
            log.error("删除文件的时候出错,文件路径{},msg:{}",path.getName(),e.getMessage());
            e.printStackTrace();
            DoopfsException.cast(CommonError.UNKOWN_ERROR);
        }
        return tag;
    }

    @Override
    public void upload(String localPath, String hadoopPath) throws DoopfsException {
        upload(new Path(localPath),new Path(hadoopPath));
    }

    @Override
    public void upload(Path localFile, Path hadoopFile) throws DoopfsException {
        try {
            fileSystem.copyFromLocalFile(localFile, hadoopFile);
        } catch (IOException e) {
            log.error("上传文件出错,文件路径{},msg:{}",hadoopFile.getName(),e.getMessage());
            e.printStackTrace();
            DoopfsException.cast(CommonError.UNKOWN_ERROR);
        }
    }

    @Override
    public void downLoad(String hadoopPath, String localPath) throws DoopfsException {
        downLoad(new Path(hadoopPath),new Path(localPath));
    }

    @Override
    public void downLoad(Path hadoopFile, Path localFile) throws DoopfsException {
        /*
         * 第一个参数控制下载完成后是否删除源文件,默认是 true,即删除;
         * 最后一个参数表示是否将 RawLocalFileSystem 用作本地文件系统;
         * RawLocalFileSystem 默认为 false,通常情况下可以不设置,
         * 但如果你在执行时候抛出 NullPointerException 异常,则代表你的文件系统与程序可能存在不兼容的情况 (window 下常见),
         * 此时可以将 RawLocalFileSystem 设置为 true
         */
        try {
            fileSystem.copyToLocalFile(false, hadoopFile, localFile, true);
        } catch (IOException e) {
            log.error("下载文件出错,文件路径{},msg:{}",hadoopFile.getName(),e.getMessage());
            e.printStackTrace();
            DoopfsException.cast(CommonError.UNKOWN_ERROR);
        }
    }

    @Override
    public FSDataInputStream getHadoopFileStream(String hadoopPath) throws DoopfsException {
        return getHadoopFileStream(new Path(hadoopPath));
    }

    @Override
    public FSDataInputStream getHadoopFileStream(Path hadoopFile) throws DoopfsException {
        FSDataInputStream stream = null;
        try {
            stream = fileSystem.open(hadoopFile);
        } catch (IOException e) {
            log.error("打开输入文件流出错出错,文件路径{},msg:{}",hadoopFile.getName(),e.getMessage());
            e.printStackTrace();
            DoopfsException.cast(CommonError.UNKOWN_ERROR);
        }
        return stream;
    }
}
